package com.tedu;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.List;

public class MyConsumer {

    @Test
    public void consumer() throws Exception {
        //1.创建消费者对象
        //1.1 创建一个消费者
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        //1.2 指定nameserver的连接地址
        defaultMQPushConsumer.setNamesrvAddr("localhost:9876");
        //1.3 指定消费者组
        defaultMQPushConsumer.setConsumerGroup("test-consumer01");
        //1.4设置主题      参数1：主题名称       参数2：过滤器
        defaultMQPushConsumer.subscribe("test-01-topic","靓女");


        //2.创建监听器，实现消费逻辑
        /**
         * 2.1 通过匿名内部类创建监听器实例对象
         * 消费者每拿到一个消息便会执行consumeMessage方法
         * 参数1：接收到的消息
         * 参数2：包含的是消费者信息
         */
        MessageListenerConcurrently messageListenerConcurrently = new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println(list);
                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();
                String msg = new String(body);
                System.out.println("消费的内容是："+msg);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        };
        //2.2 给消费者配置监听器
        defaultMQPushConsumer.setMessageListener(messageListenerConcurrently);
        //3.建立连接
        defaultMQPushConsumer.start();
        while (true);
    }
}
